package com.dominos.voldemortstress;

import java.util.Date;

import org.apache.log4j.Logger;

public class WriteThread implements Runnable {

	Logger logger = Logger.getLogger(this.getClass());

	private byte[] writeData;
	private VoldemortStoreClient storeClient;
	boolean done = false;
	boolean quit = false;
	int writeCount = 0;
	int errorCount = 0;
	long duration = 0;
	int fixedNumberOfWrites;
	int warmupWriteCount = 0;
	int numberofWarmupWrites = 0;
	private Object synchronizedForStats = new Object();

	private int writeThreadIndex;

	public WriteThread(VoldemortStoreClient voldemortStoreClient,
			int writeThreadIndex, byte[] writeData) {
		this(voldemortStoreClient, writeThreadIndex, writeData,
				Integer.MAX_VALUE, 0);
	}

	public WriteThread(VoldemortStoreClient voldemortStoreClient,
			int writeThreadIndex, byte[] writeData, int fixedNumberOfWrites) {
		this(voldemortStoreClient, writeThreadIndex, writeData,
				fixedNumberOfWrites, 0);
	}

	public WriteThread(VoldemortStoreClient voldemortStoreClient,
			int writeThreadIndex, byte[] writeData, int fixedNumberOfWrites,
			int numberOfWarmupWrites) {
		this.writeThreadIndex = writeThreadIndex;
		this.storeClient = voldemortStoreClient;
		this.writeData = writeData;
		this.fixedNumberOfWrites = fixedNumberOfWrites;
		this.numberofWarmupWrites = numberOfWarmupWrites;
	}

	@Override
	public void run() {
		quit = false;
		while (!done) {
			callPut();

		}
		quit = true;

	}

	void callPut() {
		synchronized (synchronizedForStats) {
			long startTime = new Date().getTime();
			try {
				storeClient.put(this.keyConstructor(), this.writeData);

			} catch (Exception e) {
				errorCount++;
				logger.error("Exception on Put:  " + e.getMessage());
			} finally {
				long endTime = new Date().getTime();
				long difference = endTime - startTime;
				duration += difference;
				if (this.warmupWriteCount < this.numberofWarmupWrites)
					this.warmupWriteCount++;
				else {
					writeCount++;
					if (writeCount >= this.fixedNumberOfWrites)
						done = true;
				}
			}
		}
	}

	private byte[] keyConstructor() {
		byte[] returnValue = new byte[16];

		for (int i = 0; i < 16; i++) {
			returnValue[i] = (byte) writeThreadIndex;
		}

		return returnValue;
	}

	public void quit() {
		done = true;
	}

	public int getWriteCount() {
		synchronized (synchronizedForStats) {
			return writeCount;
		}
	}

	public int getErrorCount() {
		synchronized (synchronizedForStats) {
			return errorCount;
		}
	}

	public long getDuration() {
		synchronized (synchronizedForStats) {

			return duration;
		}
	}

	public boolean isQuit() {
		return quit;
	}

	public int getWarmupWriteCount() {
		return numberofWarmupWrites;
	}

}
